/* Copyright [2023] <gyl> */
#include <vector>
#include "sql/operator/aggr_physical_operator.h"

RC AggrPhysicalOperator::open(Trx *trx) {
  if (children_.size() > 1) {
    LOG_WARN("aggregate operator must has one child");
    return RC::INTERNAL;
  }
  return children_[0]->open(trx);
}

// 当前不支持NULL类型，故COUNT(*)和COUNT(某一字段)的结果是一样的
void count(Tuple *tuple, const char *relation_name, const char *attr_name,
           Value &value, int &cnt) {
  if (strcmp(attr_name, "*") == 0) {
    value.set_int(cnt + 1);
  } else {
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), value);
    if (value.attr_type() == NULLS) {
      return;
    } else {
      value.set_int(cnt + 1);
    }
  }
  cnt++;
}

void max(Tuple *tuple, const char *relation_name, const char *attr_name,
         Value &value, int &cnt) {
  if (cnt == 0) {
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), value);
    if (value.attr_type() == NULLS) {
      return;
    }
  } else {
    Value tmp;
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), tmp);
    if (tmp.attr_type() == NULLS) {
      return;
    }
    switch (value.attr_type()) {
      case DATES:
        if (value.get_date() < tmp.get_date()) {
          value.set_date(tmp.get_date());
        }
        break;
      case INTS:
        if (value.get_int() < tmp.get_int()) {
          value.set_int(tmp.get_int());
        }
        break;

      case FLOATS:
        if (value.get_float() < tmp.get_float()) {
          value.set_float(tmp.get_float());
        }
        break;

      case CHARS:
        if (value.get_string() < tmp.get_string()) {
          value.set_string(tmp.get_string().c_str());
        }
        break;
      default:
        break;
    }
  }
  cnt++;
}

void min(Tuple *tuple, const char *relation_name, const char *attr_name,
         Value &value, int &cnt) {
  if (cnt == 0) {
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), value);
    if (value.attr_type() == NULLS) {
      return;
    }
  } else {
    Value tmp;
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), tmp);
    if (tmp.attr_type() == NULLS) {
      return;
    }
    switch (value.attr_type()) {
      case DATES: {
        if (value.get_date() > tmp.get_date()) {
          value.set_date(tmp.get_date());
        }
        break;
      }
      case INTS:
        if (value.get_int() > tmp.get_int()) {
          value.set_int(tmp.get_int());
        }
        break;
      case FLOATS:
        if (value.get_float() > tmp.get_float()) {
          value.set_float(tmp.get_float());
        }
        break;
      case CHARS:
        if (value.get_string() > tmp.get_string()) {
          value.set_string(tmp.get_string().c_str());
        }
        break;
      default:
        break;
    }
  }
  cnt++;
}

void sum(Tuple *tuple, const char *relation_name, const char *attr_name,
         Value &value, int &cnt) {
  if (cnt == 0) {
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), value);
    if (value.attr_type() == NULLS) {
      return;
    }
  } else {
    Value tmp;
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), tmp);
    if (tmp.attr_type() == NULLS) {
      return;
    }
    switch (value.attr_type()) {
      case INTS:
        value.set_int(value.get_int() + tmp.get_int());
        break;
      case FLOATS:
        value.set_float(value.get_float() + tmp.get_float());
        break;
      default:
        break;
    }
  }
  cnt++;
}

void avg(Tuple *tuple, const char *relation_name, const char *attr_name,
         Value &value, int &cnt) {
  if (cnt == 0) {
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), value);
    if (value.attr_type() == NULLS) {
      return;
    }
    switch (value.attr_type()) {
      case INTS:
        value.set_float(value.get_int());
        break;
      case FLOATS:
        value.set_float(value.get_float());
        break;
    }
    value.set_type(FLOATS);
  } else {
    Value tmp;
    tuple->find_cell(TupleCellSpec(relation_name, attr_name), tmp);
    if (tmp.attr_type() == NULLS) {
      return;
    }
    switch (tmp.attr_type()) {
      case INTS:
        value.set_float((value.get_float() * cnt + tmp.get_int()) /
                        (cnt + 1.0));
        break;
      case FLOATS:
        value.set_float((value.get_float() * cnt + tmp.get_float()) /
                        (cnt + 1.0));
        break;
      default:
        break;
    }
  }
  cnt++;
}

void make_record(int value_num, const Value *values, int record_size,
                 std::vector<FieldMeta> field, Record &record) {
  // 复制所有字段的值
  char *record_data = reinterpret_cast<char *>(malloc(record_size));
  int *null_map = reinterpret_cast<int *>(malloc(NULLMAP_LENGTH));
  *null_map = 0;  // 记录null值

  for (int i = 0; i < value_num; i++) {
    const Value &value = values[i];
    size_t copy_len = field[i].len();
    if (field[i].nullable() && value.attr_type() == NULLS) {
      *null_map |= 1 << i;
      continue;
    }
    if (field[i].type() == CHARS) {
      const size_t data_len = value.length();
      if (copy_len > data_len) {
        copy_len = data_len + 1;
      }
    }
    memcpy(record_data + field[i].offset(), value.data(), copy_len);
  }
  memcpy(record_data + NULLMAP_OFFSET, reinterpret_cast<void *>(null_map),
         NULLMAP_LENGTH);
  record.set_data_owner(record_data, record_size);
}

RC AggrPhysicalOperator::next() {
  RC rc = RC::SUCCESS;
  PhysicalOperator *oper = children_.front().get();
  // corresponding with aggregations' result.
  // note: DO NOT USE the pointer of tuple, because the record's data in it has
  // the same address, once we get a new tuple, the data of all tuples in the
  // vector will be OVERRIDDEN.
  Value *values = new Value[aggregations_.size()];
  bool exist = false;        // 是否找到了tuple
  static bool flag = false;  // 是否已经进入过此函数
  // do some initialization
  for (int i = 0; i < aggregations_.size(); i++) {
    if (aggregations_[i].aggregation == COUNT_OP) {
      values[i].set_int(0);
    } else {
      values[i].set_null();
    }
  }

  int *cnt = new int[aggregations_.size()];
  for (int i = 0; i < aggregations_.size(); i++) {
    cnt[i] = 0;
  }
  while (RC::SUCCESS == (rc = oper->next())) {
    exist = true;
    Tuple *tuple = oper->current_tuple();
    if (nullptr == tuple) {
      rc = RC::INTERNAL;
      LOG_WARN("failed to get tuple from operator");
      break;
    }
    for (int i = 0; i < aggregations_.size(); i++) {
      const char *relation_name =
          aggregations_[i].attribute.relation_name.c_str();
      const char *attr_name = aggregations_[i].attribute.attribute_name.c_str();
      Value tmp;
      tuple->find_cell(TupleCellSpec(relation_name, attr_name), tmp);
      if (tmp.attr_type() == NULLS) {
        continue;
      }
      switch (aggregations_[i].aggregation) {
        case MAX_OP:
          max(tuple, relation_name, attr_name, values[i], cnt[i]);
          break;
        case MIN_OP:
          min(tuple, relation_name, attr_name, values[i], cnt[i]);
          break;
        case COUNT_OP:
          count(tuple, relation_name, attr_name, values[i], cnt[i]);
          break;
        case SUM_OP:
          sum(tuple, relation_name, attr_name, values[i], cnt[i]);
          break;
        case AVG_OP:
          avg(tuple, relation_name, attr_name, values[i], cnt[i]);
          break;
        default:
          return RC::INTERNAL;  // should not reach here.
      }
    }
  }

  // here we use a pointer. if not, when the function returns, the RowTuple will
  // be destroyed, which will cause the next destruction gets an invalid
  // address. amazing.
  RowTuple *row_tuple = new RowTuple();
  Record *record = new Record();
  std::vector<FieldMeta> *field_meta = new std::vector<FieldMeta>;
  int len = aggregations_.size();
  for (int i = 0; i < len; i++) {
    AttrType type = values[i].attr_type();
    if (type == UNDEFINED) {
      return RC::INTERNAL;
    }
    field_meta->push_back(FieldMeta(to_string(i).c_str(), type,
                                    i * 4 + NULLMAP_LENGTH, 4, true, 1));
  }
  make_record(len, values, len * 4 + NULLMAP_LENGTH, *field_meta, *record);

  Table *table = new Table();
  row_tuple->set_record(record);
  row_tuple->set_schema(NULL, field_meta);
  tuple_ = *row_tuple;
  // the returned rc is always RECORD_EOF due to the while statement,
  // so we check exist to modify the result of rc;
  // 如果找到了一些tuple并且返回结果为RECORD_EOF或者没有找到tuple但却是第一次进入此函数，这两种情况都应该输出
  if ((exist && rc == RC::RECORD_EOF) || (!exist && !flag)) {
    flag = true;  // 更改flag标记为进入过此函数
    return RC::SUCCESS;
  }
  flag = false;  // 更改flag标记为未进入过，以等待下一次查询
  return rc;
}

RC AggrPhysicalOperator::close() {
  if (!children_.empty()) {
    children_[0]->close();
  }
  return RC::SUCCESS;
}

Tuple *AggrPhysicalOperator::current_tuple() { return &tuple_; }
